
package com.huaweicloud.saaslanding.install.applicationservice.impl;

import com.huaweicloud.saaslanding.basic.constant.KafkaConstant;
import com.huaweicloud.saaslanding.basic.database.mybatis.conditions.query.QueryWrap;
import com.huaweicloud.saaslanding.basic.utils.BeanPlusUtil;
import com.huaweicloud.saaslanding.basic.utils.EncryptionUtil;
import com.huaweicloud.saaslanding.basic.utils.ResourceNamingUtil;
import com.huaweicloud.saaslanding.dao.common.entity.cloudservice.RocketMQInfo;
import com.huaweicloud.saaslanding.dao.common.entity.security.WorkSecretKey;
import com.huaweicloud.saaslanding.dao.common.entity.tenant.Tenant;
import com.huaweicloud.saaslanding.dao.common.repository.cloudservice.RocketMQInfoBaseMapper;
import com.huaweicloud.saaslanding.dao.common.repository.security.WorkSecretKeyBaseMapper;
import com.huaweicloud.saaslanding.dao.common.repository.tenant.TenantBaseMapper;
import com.huaweicloud.saaslanding.dao.common.view.RocketMQAuthInfo;
import com.huaweicloud.saaslanding.huawei.rocketmq.api.RocketMQApi;
import com.huaweicloud.saaslanding.huawei.rocketmq.constant.RocketMQConstant;
import com.huaweicloud.saaslanding.huawei.rocketmq.dto.RocketMQAuthDto;
import com.huaweicloud.saaslanding.huawei.rocketmq.enumeration.RocketMQPurposeTypeEnum;
import com.huaweicloud.saaslanding.install.applicationservice.RocketMQService;
import com.huaweicloud.saaslanding.install.properties.RocketMQ;
import com.huaweicloud.saaslanding.install.repository.RdsRepository;
import com.huaweicloud.sdk.rocketmq.v2.model.ListRocketInstanceTopicsResponse;
import com.huaweicloud.sdk.rocketmq.v2.model.Topic;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * 功能描述
 *
 * @since 2023-03-10
 */
@Slf4j
@Service
@ConditionalOnExpression("${saaslanding-install.lts.enabled:false} && 'RocketMQ'.equalsIgnoreCase('${saaslanding-install.lts.consumer}')")
public class RocketMQServiceImpl implements RocketMQService {
    private static final String TABLE_NAME = "_rocketmq_info";

    @Resource
    private WorkSecretKeyBaseMapper workSecretKeyBaseMapper;

    @Resource
    private RocketMQInfoBaseMapper rocketMQInfoBaseMapper;

    @Resource
    private TenantBaseMapper tenantBaseMapper;

    @Resource
    private RdsRepository rdsRepository;

    @Resource
    private RocketMQ rocketMQ;

    @Value("${saaslanding-install.database.connection-info.db-name}")
    private String database;

    /**
     * 初始化RocketMQ信息
     */
    @Override
    public void initRocketMQ() {
        log.info("开始创建RocketMQInfo表并初始化数据...");
        rdsRepository.createRocketMQInfoTable(database, ResourceNamingUtil.getPrefixCode());

        RocketMQInfo rocketMQInfo = BeanPlusUtil.toBean(rocketMQ, RocketMQInfo.class)
            .setDescription(RocketMQPurposeTypeEnum.OP_LOG_TRANSFER.getDescription())
            .setPurposeType(RocketMQPurposeTypeEnum.OP_LOG_TRANSFER.getType());
        rocketMQInfoBaseMapper.insert(rocketMQInfo);

        log.info("初始化RocketMQ信息成功.");
    }

    /**
     * 删除RocketMQ的主题和消费者组
     */
    @Override
    public void deleteRocketMQ() {
        if (!rdsRepository.isTableExist(database, ResourceNamingUtil.getPrefixCode().concat(TABLE_NAME))) {
            log.info("数据库表未创建，跳过删除资源.");
            return;
        }

        // 主题名称是租户code，删除租户时可能没有删除主题，查询所有租户
        List<String> tenantList =
            tenantBaseMapper.selectList(new QueryWrap<Tenant>().select("code")).stream().map(Tenant::getCode).toList();
        if (tenantList.isEmpty()) {
            log.info("租户列表为空，跳过删除RocketMQ主题和消费组...");
            return;
        }

        log.info("开始删除RocketMQ主题和消费组...");
        List<String> topicNameList;
        RocketMQAuthDto authDto = getRocketMQAuthDto();
        if (authDto == null) {
            log.warn("未找到RocketMQ连接信息，可能是RocketMQ实例与IAM账号处于不同region！！！\n请手动前往RocketMQ删除主题及消费组.");
            return;
        }

        int offset = 0;
        try {
            ListRocketInstanceTopicsResponse response =
                RocketMQApi.getTopicList(authDto, RocketMQConstant.DEFAULT_LIMIT, offset);
            Integer total = response.getTotal();
            if (total <= 0) {
                log.info("RocketMQ中的主题数量为空，跳过删除主题和消费者...");
                return;
            }

            topicNameList = new ArrayList<>(response.getTopics().stream().map(Topic::getName).toList());
            offset += RocketMQConstant.DEFAULT_LIMIT;
            while (total > offset) {
                response = RocketMQApi.getTopicList(authDto, RocketMQConstant.DEFAULT_LIMIT, offset);
                topicNameList.addAll(response.getTopics().stream().map(Topic::getName).toList());
                offset += RocketMQConstant.DEFAULT_LIMIT;
            }

            topicNameList = topicNameList.stream().filter(tenantList::contains).toList();
            RocketMQApi.batchDeleteTopics(authDto, topicNameList);
            log.info("删除RocketMQ主题成功.");
        } catch (Exception e) {
            log.error("删除RocketMQ主题失败，可能是连接信息出错，请手动前往RocketMQ删除主题及消费组！！！\n错误信息：{}", e.getLocalizedMessage());
            return;
        }

        try {
            // 默认每个主题只有一个消费组，且没有被手动删除，如果考虑到手动删除消费组情况，需要根据每个主题查询对应的消费组
            List<String> consumerGroups =
                topicNameList.stream().map(KafkaConstant.DEFAULT_CONSUMER_PREFIX::concat).toList();
            RocketMQApi.batchDeleteConsumerGroups(authDto, consumerGroups);
            log.info("删除RocketMQ消费组成功.");
        } catch (Exception e) {
            log.error("删除RocketMQ消费组失败，可能是消费组错误，请手动前往RocketMQ删除消费组！！！\n错误信息：{}", e.getLocalizedMessage());
        }
    }

    /**
     * 获取RocketMQ实例连接信息
     *
     * @return RocketMQ实例连接信息
     */
    private RocketMQAuthDto getRocketMQAuthDto() {
        WorkSecretKey workSecretKey = workSecretKeyBaseMapper.getPlatformSalt();
        if (workSecretKey == null) {
            log.warn("获取盐失败，RocketMQ连接信息无法解密...");
            return null;
        }
        String pwdSalt = EncryptionUtil.decrypt(workSecretKey.getPwdSalt(), workSecretKey.getPassword());

        RocketMQAuthInfo authInfo =
            rocketMQInfoBaseMapper.getRocketMQInstanceByType(RocketMQPurposeTypeEnum.OP_LOG_TRANSFER.getType());
        if (authInfo == null) {
            return null;
        }

        RocketMQAuthDto authDto = BeanPlusUtil.toBean(authInfo, RocketMQAuthDto.class);
        return authDto.setAk(EncryptionUtil.decrypt(pwdSalt, authInfo.getAk()))
            .setSk(EncryptionUtil.decrypt(pwdSalt, authInfo.getSk()));
    }
}
